第1世代Pub/SubトリガーのGoogle Cloud Functionsを第2世代のHTTPトリガーに書き換えてみた

Pub/Subトリガーで実装したGoogle Cloud Functionsを新しい第2世代に作り替える。トリガーはタイムアウトの長いHTTPトリガーに変更し、Pub/Subから起動するように設定する。

#Google Cloud Functions

#Google Cloud (GCP)

ikeda

2022.11.11

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

データアナリティクス事業本部、池田です。この記事は夏のうちに書きたかったのですが、とっくに終わってしまいましたね。

Google Cloud Functions の第二世代（2nd gen）が今夏 GA になりました。

【 Cloud Functions（第 2 世代）の一般提供を開始: より多くのイベント、コンピューティング、コントロールを提供】

世代間の比較も公式ガイドにあります。
【 Cloud Functions バージョンの比較＞比較表】

上記ガイドの中でも、 移行の機能は提供予定（執筆時点）とはなっていますが、今回は自分でPythonコードやデプロイのスクリプトを書き換えて移行してみます。

Cloud Functions（第 2 世代）で近日提供予定
…
・Cloud Functions（第 1 世代）の関数を（第 2 世代）に移行して、新しい機能を使用できるようにします。

以前に以下のブログで実装した、 Cloud Pub/Sub をトリガーに設定した第1世代のCloud Functions（Python）を書き換えます。

Cloud FunctionsでGoogle Cloud Python Loggingライブラリを使ってみる（初心者向け）

（当時はロギングの紹介で実装しました。）

ただし、Pub/Subトリガーでなく、HTTPトリガーに変更します。

※そのままPub/Subトリガーで書き換える場合は、公式のチュートリアルが参考になると思います。

HTTPトリガーに変更する理由は、上記チュートリアルにも記載がある通り複数サブスクライブ登録できるようになることと…

注: HTTP トリガー関数を使用して Pub/Sub push サブスクリプションをリッスンできます。HTTP 関数を使用すれば、1 つの関数で複数の Pub/Sub トピックに登録できます。

第2世代ではPub/Subトリガー（イベントトリガーに含まれる）のタイムアウトは9分のままなのに対し、 HTTPトリガーは60分に延長されているためです。

マイグレーション

公式のHTTPトリガーのチュートリアルを参考に進めました。

Cloud Functions実装変更

デプロイの際に必要なAPIを有効化します。 Cloud Functions・Cloud Build・Artifact Registry・Cloud Run・Logging・Pub/SubのAPIが必要とのこと。
管理上問題無ければ、↓のようにコンソールで軽く作成を始めてまとめて有効化してしまうのが楽そうでした。

変更前後のコードを記載していきます。

import base64
import os
import json
from logging import getLogger, DEBUG
import google.cloud.logging

logging_client = google.cloud.logging.Client()
logging_client.setup_logging()
logger = getLogger(__name__)
logger.setLevel(DEBUG)

def sample_logging_fn(event, context):
    pubsub_message = base64.b64decode(event["data"]).decode("utf-8")
    logger.info(f"pubsub_message: {pubsub_message}")

    logger.debug(json.dumps(event))
    logger.debug(event)

    logger.warning(f"env: {os.getenv('ENV_VAR')}")

    try:
        logger.info("info at try")
        _ = 1 / 0
    except Exception:
        logger.error("error at except")
        raise
    finally:
        logger.debug("debug at finally")

↓↓↓↓

import base64
import os
import functions_framework
from logging import getLogger, DEBUG
import google.cloud.logging

logging_client = google.cloud.logging.Client()
logging_client.setup_logging()
logger = getLogger(__name__)
logger.setLevel(DEBUG)

@functions_framework.http
def sample_logging_fn_gen2(request):
    pubsub_message = base64.b64decode(request.get_json(silent=True)["message"]["data"]).decode("utf-8")
    logger.info(f"pubsub_message: {pubsub_message}")

    logger.debug(request)

    logger.warning(f"env: {os.getenv('ENV_VAR')}")

    try:
        logger.info("info at try")
        _ = 1 / 0
    except Exception:
        logger.error("error at except")
        # raise
    finally:
        logger.debug("debug at finally")

    return "OK"

変更後のコードを元にポイントだけ説明していきます。

import functions_framework

@functions_framework.http

Functions Framework というものらしいです。ローカルで開発しやすかったり移植性を高めたりするものだと理解しています。たぶん。
これは第1世代でも利用できるようなのですが、 第2世代からコンソールでコードを作成した時に記述されるようになったので追加しました。

def sample_logging_fn_gen2(request):
    pubsub_message = base64.b64decode(request.get_json(silent=True)["message"]["data"]).decode("utf-8")

第1世代と第2世代、イベントトリガーとHTTPトリガーでは、エントリポイントが受け取る引数や、発行されたPub/Subメッセージ部分の取り出し方が変わります。
引数は Request のオブジェクトです。

    except Exception:
        logger.error("error at except")
        # raise
    finally:
        logger.debug("debug at finally")

    return "OK"

第1世代Pub/Subトリガーでは明示的なreturnが不要でしたが、 HTTPトリガーではreturnが必要なようでした。
また、第1世代Pub/Subトリガーでエラーが発生しても Pub/Subサブスクリプション上は配信成功扱いになったのですが、 HTTPトリガーでは配信エラーとなりサブスクリプションが再実行を繰り返してしまうので、今回のサンプルはエラーを握り潰すようにしました。

google.cloud.logging>=3.0.0

↓↓↓↓

functions-framework>=3.0
google.cloud.logging>=3.2.0

前述の追加した functions-framework を追記。

gcloud functions deploy sample_logging_fn \
--region asia-northeast1 \
--runtime python39 \
--trigger-topic  sample_logging_topic \
--timeout 30 \
--memory 128MB \
--set-env-vars ENV_VAR=環境変数からの値

↓↓↓↓

gcloud functions deploy sample-logging-fn-gen2 \
--gen2 \
--region asia-northeast1 \
--runtime python310 \
--entry-point=sample_logging_fn_gen2 \
--trigger-http \
--allow-unauthenticated \
--timeout 30 \
--memory 128MiB \
--set-env-vars ENV_VAR=環境変数からの値

公式のコマンドのリファレンスを参考に変更しました。

gcloud functions deploy sample-logging-fn-gen2 \

--entry-point=sample_logging_fn_gen2 \

第2世代ではCloud Functionsの名称に _ が使えなくなったみたいです。 ↓こんな感じのデプロイエラーになりました。
ERROR: (gcloud.functions.deploy) INVALID_ARGUMENT: Could not create Cloud Run service sample_logging_fn_gen2. metadata.name: Resource name must use only lowercase letters, numbers and '-'. Must begin with a letter and cannot end with a '-'. Maximum length is 63 characters.
更にその影響で、名称とエントリポイントの関数名が違ってしまうので、 エントリポイントを明示的に指定するようにしました。

--gen2 \

第2世代であることを示すためのオプションです。

--trigger-http \
--allow-unauthenticated \

HTTPトリガーの設定です。
--allow-unauthenticated は「未認証の呼び出しを許可」する設定です。今回はPub/Sub側の設定を簡略化するために有効にしましたが、利用には注意が必要です。

--memory 128MiB \

単位が MB のままだとデプロイエラーになりました。罠！

このコマンド/シェルでデプロイを行うと、↓のようになります。

コンソールや後述のコマンドで呼び出すためのURLが取得できます。
URLの末尾は .run.app になっています。（ cloudfunctions.net URLもサポートされる予定とのこと。）

Cloud Pub/Sub側の設定

Pub/Subトピックは以下のコマンドで作成済みのものを利用します。
gcloud pubsub topics create sample_logging_topic

コマンドで前節で作成したCloud Functions関数の呼び出しURLを取得し、トピックへサブスクライブ登録します。

PUSH_ENDPOINT_URI=`gcloud functions describe sample-logging-fn-gen2 --gen2 --region asia-northeast1 --format="value(serviceConfig.uri)"`

gcloud pubsub subscriptions create gcf-sample-logging-fn-gen2-subscription \
--topic=sample_logging_topic \
--expiration-period=never \
--push-endpoint=$PUSH_ENDPOINT_URI

先述の通りHTTPトリガーのCloud Functions関数内で例外が起こるとサブスクリプションが再実行を繰り返してしまうので、関数の内容によってはデッドレタリングの設定などが必要そうです。